package com.zy.demokafka.consumer;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import scala.Array;

import java.util.Arrays;
import java.util.Properties;

public class Myconsumery {
    public static void main(String[] args) {

        Properties props = new Properties();
        //给配置信息负值  测试2
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.0.101:9092");
        //开启自动提交
//        props.put("enable.auto.commit", "true");
        props.put("enable.auto.commit", "false");
        props.put("auto.commit.interval.ms", "1000");
        props.put("key.deserializer",
                "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer",
                "org.apache.kafka.common.serialization.StringDeserializer");
        //设置消费者组
        props.put("group.id", "test11");
        //重置消费者offset
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(props);
        kafkaConsumer.subscribe(Arrays.asList("first","sun"));

        while (true){  //保证长时间拉取生产者数据
            //kafka是拉取数据设置的轮训时间
            ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(100);
            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                System.out.println(consumerRecord.key()+"==="+consumerRecord.value());
            }
        }

    }
}
